package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import com.bw.gmall.realtime.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

//加购事实表
public class DwdTradeCartAdd {
    public static void main(String[] args) throws Exception{
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //设置kafka主题的分区数
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        //1.1 开启CheckPoint
        //env.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE);
        //env.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);
        //env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000L));
        //1.2 设置状态后端
        //env.setStateBackend(new HashMapStateBackend());
        //env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/211126/ck");
        //System.setProperty("HADOOP_USER_NAME", "bw");

        //使用ddl方式读取kafka数据
        tEnv.executeSql(MyKafkaUtil.getTopicDb("cart_add_250915"));
        //topic_db
        //过滤出加购数据
        Table cartAddTable = tEnv.sqlQuery("select" +
                " `data`['id'] id, " +
                " `data`['user_id'] user_id, " +
                " `data`['sku_id'] sku_id, " +
                " `data`['cart_price'] cart_price, " +
                // 修复if条件判断和类型转换
                " if(`type`='insert', `data`['sku_num'], cast(cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int) as string)) sku_num, " +

                " `data`['sku_name'] sku_name, " +
                " `data`['is_checked'] is_checked, " +
                " `data`['create_time'] create_time, " +

                " `data`['operate_time'] operate_time, " +
                " `data`['is_ordered'] is_ordered, " +
                " `data`['order_time'] order_time, " +
                " `data`['source_type'] source_type, " +
                " `data`['source_id'] source_id, " +
                " pt " +
                " from topic_db " +

                " where `database`='gmall' " +
                " and `table`='cart_info' " +
                " and (`type`='insert' " +
                " or (`type`='update' " +
                " and `old`['sku_num'] is not null " +
                // 修复字段名错误（date改为data）
                " and cast(`data`['sku_num'] as int) > cast(`old`['sku_num'] as int)))");
        //将加购表转换为流并打印测试
        tEnv.createTemporaryView("cart_info_table", cartAddTable);
        //购物车表
//        tEnv.toAppendStream(cartAddTable, Row.class).print();
//        tEnv.sqlQuery("select * from cart_info_table").execute().print();

        //读取mysql的base_dic 表 作为LookUp表
        tEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());

        //关联两张表
        //TODO 5.关联两张表
        Table cartAddWithDicTable = tEnv.sqlQuery("" +
                "select " +
                "    ci.id, " +
                "    ci.user_id, " +
                "    ci.sku_id, " +
                "    ci.cart_price, " +
                "    ci.sku_num, " +
                "    ci.sku_name, " +
                "    ci.is_checked, " +
                "    ci.create_time, " +
                "    ci.operate_time, " +
                "    ci.is_ordered, " +
                "    ci.order_time, " +
                "    ci.source_type source_type_id, " +
                "    dic.dic_name source_type_name, " +
                "    ci.source_id " +
                "from cart_info_table ci " +
                "join base_dic FOR SYSTEM_TIME AS OF ci.pt as dic " +
                "on ci.source_type = dic.dic_code");
        tEnv.createTemporaryView("cart_add_dic_table", cartAddWithDicTable);

        //TODO 6.使用DDL方式创建加购事实表
        tEnv.executeSql("" +
                "create table dwd_cart_add( " +
                "    `id` STRING, " +
                "    `user_id` STRING, " +
                "    `sku_id` STRING, " +
                "    `cart_price` STRING, " +
                "    `sku_num` STRING, " +
                "    `sku_name` STRING, " +
                "    `is_checked` STRING, " +
                "    `create_time` STRING, " +
                "    `operate_time` STRING, " +
                "    `is_ordered` STRING, " +
                "    `order_time` STRING, " +
                "    `source_type_id` STRING, " +
                "    `source_type_name` STRING, " +
                "    `source_id` STRING " +
                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_trade_cart_add"));

        //TODO 7.将数据写出
        //tableEnv.executeSql("insert into dwd_cart_add select * from " + cartAddWithDicTable);
        tEnv.executeSql("insert into dwd_cart_add select * from cart_add_dic_table");
        //.print();

        //TODO 8.启动任务
        env.execute("DwdTradeCartAdd");









    }
}
